package com.ayz.gateway.center.domain.service.remoting;


import com.ayz.gateway.api.request.GatewayServiceSubscriber;
import com.ayz.gateway.api.response.GatewayCommonResult;
import com.ayz.gateway.api.server.SubscribeGatewayGroupServiceGrpc;
import com.ayz.gateway.center.domain.repository.IApiRepository;
import com.ayz.gateway.center.domain.repository.IGatewayServiceRepository;
import com.ayz.gateway.center.infrastructure.notify.ApiDataChangeEvent;
import com.ayz.gateway.center.infrastructure.notify.GatewayServiceLogoutEvent;
import com.ayz.gateway.common.constants.GatewayServiceData;
import com.ayz.gateway.common.notify.ApplicationEvent;
import com.ayz.gateway.common.notify.NotifyCenter;
import com.ayz.gateway.common.result.ResponseCode;
import io.grpc.stub.StreamObserver;
import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.LinkedBlockingQueue;


/**
 * @ClassName SubscribeGatewayGroupServiceImpl
 * @Description TODO: 网关服务向网关注册中心订阅某一个网关业务组
 * @Author 快乐的星球
 * @Date 2023/5/28 15:05
 * @Version 1.0
 **/
@Getter
@Setter
public class SubscribeGatewayGroupServiceImpl extends SubscribeGatewayGroupServiceGrpc.SubscribeGatewayGroupServiceImplBase {
    /**
     * API网关注册中心API的核心存储功能类
     */
    private IApiRepository apiRepository;
    /**
     * 网关服务注册中心
     */
    private IGatewayServiceRepository gatewayServiceRepository;

    public SubscribeGatewayGroupServiceImpl(IApiRepository apiRepository, IGatewayServiceRepository gatewayServiceRepository) {
        this.apiRepository = apiRepository;
        this.gatewayServiceRepository = gatewayServiceRepository;
    }
    @Override
    public void subscribeGatewayGroup(GatewayServiceSubscriber.SubscribeGatewayGroupRequest request, StreamObserver<GatewayCommonResult.GatewayResult> responseObserver) {
        //1:解析请求
        String gatewayId = request.getGatewayId();
        //2:获取网关服务基础信息
        GatewayServiceData gatewayServiceData = gatewayServiceRepository.getGatewayServiceData(gatewayId);
        if(gatewayServiceData==null){
            //说明当前网关服务还没有注册
            //订阅失败
            GatewayCommonResult.GatewayResult gatewayResult = GatewayCommonResult.GatewayResult.newBuilder()
                    .setCode(ResponseCode.SUBSCRIBE_GATEWAY_GROUP_FAILED).setMsg("网关服务还未注册订阅失败!网关ID:{" + gatewayId + "}").build();
            responseObserver.onNext(gatewayResult);
            responseObserver.onCompleted();
        }else{
            //网关服务已经注册过,判断是否已经订阅过，已经订阅过则不能再进行订阅
            boolean subscribed=gatewayServiceRepository.containsGatewayServiceSubscriber(gatewayId);
            if(subscribed){
                //已经订阅过
                GatewayCommonResult.GatewayResult gatewayResult = GatewayCommonResult.GatewayResult.newBuilder()
                        .setCode(ResponseCode.SUBSCRIBE_GATEWAY_GROUP_FAILED).setMsg("网关服务已经订阅过,不允许重复订阅!网关ID:{" + gatewayId + "}").build();
                responseObserver.onNext(gatewayResult);
                responseObserver.onCompleted();
                return;
            }
            //1:创建网关服务订阅者
            com.ayz.gateway.center.infrastructure.notify.GatewayServiceSubscriber
                    gatewayServiceSubscriber=new com.ayz.gateway.center.infrastructure.notify.GatewayServiceSubscriber(gatewayServiceData);
            //2:注册订阅者
            //2.1: 向网关注册中心网关服务存储类
            gatewayServiceRepository.addGatewayServiceSubscriber(gatewayId,gatewayServiceSubscriber);
            NotifyCenter notifyCenter = NotifyCenter.getINSTANCE();
            notifyCenter.getEventMulticaster().addApplicationListener(gatewayServiceSubscriber);
            //获取事件阻塞队列
            LinkedBlockingQueue<ApplicationEvent> applicationEventBlockingQueue = gatewayServiceSubscriber.getApplicationEventBlockingQueue();
            while (true){
                try {
                    ApplicationEvent applicationEvent = applicationEventBlockingQueue.take();
                    if(applicationEvent instanceof ApiDataChangeEvent){
                        if(gatewayServiceSubscriber.isSubscribeGatewayGroup(applicationEvent)){
                            //当前网关服务订阅的网关组下API集合发生变更
                            //向网关服务推送
                            gatewayServiceSubscriber.pushApiDataChange(responseObserver,getApiRepository());
                        }
                    }else if(applicationEvent instanceof GatewayServiceLogoutEvent){
                        //网关服务注销事件
                        GatewayServiceLogoutEvent gatewayServiceLogoutEvent=(GatewayServiceLogoutEvent) applicationEvent;
                        if(gatewayServiceLogoutEvent.getGatewayId().equals(gatewayId)){
                            //当前网关服务注销
                            GatewayCommonResult.GatewayResult gatewayResult = GatewayCommonResult.GatewayResult.newBuilder().setCode(ResponseCode.GATEWAY_SERVICE_LOGOUT_EVENT)
                                    .setMsg("当前网关服务注销!").build();
                            responseObserver.onNext(gatewayResult);
                            responseObserver.onCompleted();
                            return;
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
